package cn.jbit.reactor.worker.server;

import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.SocketChannel;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * 源码学院只为培养优秀的bat程序员而生
 *
 * @author 学员张在成
 * @data 2020/8/8
 */
public class AsyncHandler implements Runnable {

        private  final Selector selector;

        private  final SelectionKey selectionKey;

        private  final SocketChannel  socketChannel;

        private ByteBuffer  readBuffer=ByteBuffer.allocate(1024);

        private  ByteBuffer sendBUffer=ByteBuffer.allocate(1024);

        private  final static  int READ=0;//读取就绪

        private  final  static  int SEND=1;//响应就绪

        private  final  static  int PROCESSING=2;//处理中

        private  int status=READ;  //所有连接完成都是从读取动作开始的

        //开启线程数为5的异步处理线程池
        private  static  final ExecutorService workers= Executors.newFixedThreadPool(5);


        public AsyncHandler(SocketChannel socketChannel,Selector selector) throws IOException {
            this.socketChannel = socketChannel;
            this.socketChannel.configureBlocking(false);
            selectionKey=socketChannel.register(selector,0);
            selectionKey.attach(this);
            selectionKey.interestOps(SelectionKey.OP_READ);
            this.selector = selector;
            this.selector.wakeup();
        }


        @Override
        public void run() {

            switch (status){
                case  READ:
                    try {
                        read();
                    } catch (IOException e) {
                        e.printStackTrace();
                    }
                    break;
                case  SEND:
                    send();
                    break;
                default:
            }
        }

        public  void read() throws IOException {
            if (selectionKey.isValid()){
                try {
                    readBuffer.clear();
                    int count=socketChannel.read(readBuffer);
                    if (count>0){
                        //将执行的状态置为PROCESSING
                        status=PROCESSING;
                        workers.execute(this::readWorker);
                    }else {
                        selectionKey.cancel();
                        socketChannel.close();
                        System.out.println("read时，连接关闭");
                    }
                }catch (IOException e){
                    System.err.println("处理read业务时发生异常!异常信息："+e.getMessage());
                    selectionKey.cancel();
                    try {
                        socketChannel.close();
                    }catch (IOException e1){
                        System.err.println("处理read业务关闭通道时发生异常！异常信息："+e1.getMessage());
                    }
                }
            }
        }
        //读入信息后的业务处理
        private void readWorker() {
            System.out.println(String.format("收到来自客户端的消息：%s",new String(readBuffer.array())));
            status=SEND;
            selectionKey.interestOps(SelectionKey.OP_WRITE);//把当前事件改为写事件

            selector.wakeup();//唤醒阻塞在select的线程
        }

         public void send(){
            if (selectionKey.isValid()){
                status=PROCESSING;//置为执行中
                workers.execute(this::sendWorker);   //异步处理
                selectionKey.interestOps(SelectionKey.OP_READ);  //重新设置为读
            }
         }

        private void sendWorker(){
                try {
                    sendBUffer.clear();
                    sendBUffer.put(String.format("我收到来自%s的信息啦：%s, 200ok;",
                            socketChannel.getRemoteAddress(),new String(readBuffer.array())).getBytes());

                    sendBUffer.flip();
                    int count=socketChannel.write(sendBUffer);
                    if (count<0){
                        selectionKey.cancel();
                        socketChannel.close();
                        System.out.println("send时---连接关闭");
                    }else {
                        //再次切换到读
                        status=READ;
                    }
                }catch (IOException e){
                    System.err.println("异步处理业务时发生异常！异常信息："+e.getMessage());
                    selectionKey.cancel();
                    try {
                        socketChannel.close();
                    }catch (IOException e1){
                        System.err.println("异步处理send业务关闭通道时发生异常! 异常信息："+e.getMessage());
                    }
                }
        }
}
